package com.hk.core.redis.locks;

import com.hk.commons.util.ArrayUtils;
import com.hk.commons.util.Lazy;
import com.hk.commons.util.ObjectUtils;
import com.hk.commons.util.SpringContextHolder;
import com.hk.core.redis.scripting.RedisScriptSource;
import lombok.SneakyThrows;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;

import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * <pre>
 * 基于 Redis set NX 与 lua 脚本实现分布式锁，支持可重入。
 * 从redis 2.6.12版本开始，SET 命令可以支持 加参数 NX/EX/PX等, 不需要再使用（SETNX, SETEX, PSETEX） 这些命令了
 *
 *   注意: redis的锁一般不要用在redis集群、哨兵中，
 *      因为多台redis集群中，会存在数据延迟问题，可能在主节点上获取锁成功，在从节点上又获取不到，就会导致锁失败。
 *
 *      redisson 有个 RedissonRedLock，锁多台redis实例，也会存在一定的问题。
 *         比如： 有5台服务器，
 *          第一次使用 1、2、3台服务器获取到了锁，但因为网络、等各种问题，第3台服务器造成对此应用不通，但第3台服务器对此应用的第二个副本又是通的
 *          那这个副本应用可以在3、4、5这3台服务器获取到锁，最后的结果就是，应用的两个不同副本的不同线程同时会获取到锁。
 * </pre>
 *
 * @author Kevin
 * @date 2019-2-20 22:08
 */
public class RedisLock implements Lock {

    /**
     * 线程睡眠时间，单位:毫秒
     */
    private static final long SLEEP_TIME = 10;

    /**
     * <pre>
     * 默认过期时间: 20 秒
     * </pre>
     */
    private static final long EXPIRE_SECONDS = 20;
    /**
     * lua 脚本内容，lua 脚本能保证原子性执行
     */
    private static final DefaultRedisScript<Long> LUA_SCRIPT;

    private static final RedisLockRenewalStrategy redisLockRenewalStrategy;

    /**
     * 非事物的 redisTemplate
     */
    private static final Lazy<StringRedisTemplate> STRING_REDIS_TEMPLATE_LAZY =
            Lazy.of(() -> SpringContextHolder.getBean("stringRedisTemplateDisabledTransactionSupport",
                    StringRedisTemplate.class));

    static {
        LUA_SCRIPT = new DefaultRedisScript<>();
//        LUA_SCRIPT.setScriptSource(new ResourceScriptSource(new ClassPathResource("META-INF/scripts/unlock.lua")));
        LUA_SCRIPT.setScriptSource(RedisScriptSource.LOCK);
        LUA_SCRIPT.setResultType(Long.class);
        redisLockRenewalStrategy = RedisLockRenewalStrategyHolder.getRenewalStrategy();
    }

    /**
     * redis Key
     */
    private final String key;

    private final String value;

    /**
     * key 过期时间，防止死锁
     */
    private final long expire;

    /**
     * key 是否续期，
     * 一般情况下，可以将 {@link #expire} 时间设置相对长一点，可以不续期，只要保证过期时间内，业务能执行完即可。
     */
    private final boolean renewal;

    /**
     * 默认过期时间 2 秒
     *
     * @param key key
     */
    public RedisLock(String key) {
        this(key, EXPIRE_SECONDS);
    }

    /**
     * 默认过期时间 2 秒
     *
     * @param key key
     */
    public RedisLock(String key, long expire) {
        this(key, expire, false);
    }

    /**
     * @param key     key
     * @param expire  key过期时间 默认过期时间 2 秒
     * @param renewal 是否对key进行续期
     */
    public RedisLock(String key, long expire, boolean renewal) {
        this.key = key;
        this.expire = expire <= 0 ? EXPIRE_SECONDS : expire;
        this.renewal = renewal;
        this.value = String.format("lock_%s_%f", key, Math.random());
    }

    /**
     * 获取锁，使用递归，无法使用优雅方式，即当一个线程释放锁后不能自动通知其它等待的客户端来获取锁，而使用递归每 10 ms 来获取一次.
     */
    @Override
    public void lock() {
        while (!tryLock()) {
            if (SLEEP_TIME >= 0) {
                try {
                    TimeUnit.NANOSECONDS.sleep(SLEEP_TIME);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }

    /**
     * 可被打断的获取 lock
     */
    @Override
    public void lockInterruptibly() throws InterruptedException {
        if (Thread.interrupted()) {//如果当前线程已中段
            throw new InterruptedException();
        }
        while (!tryLock()) {
            TimeUnit.NANOSECONDS.sleep(SLEEP_TIME);
        }
    }

    /**
     * 尝试获取锁，立即返回。如果返回  true ,加锁成功
     *
     * @return true or false
     */
    @Override
    public boolean tryLock() {
        /*
         * 只有当Key 不存在时，设置key的值，
         *  注意，redisTemplate.setIfAbsent方法，在 pipeline / transaction 的方法调用此方法时， 会返回 null，
         *  所以这里使用  stringRedisTemplateDisabledTransactionSupport 的 redisTemplate
         */
        var result = ObjectUtils.defaultIfNull(STRING_REDIS_TEMPLATE_LAZY.get().opsForValue().setIfAbsent(key, value,
                expire, TimeUnit.SECONDS), Boolean.FALSE);
        if (result && renewal && expire > 1) { //过期时间大于1秒才设置开启线程定时刷新过期时间
            registerScheduleIfRenewal();
        }
        return result;
    }

    /**
     * 初始化线程并启动
     */
    private void registerScheduleIfRenewal() {
        cancelScheduleIfRenewal();
        long period = expire > 2 ? (expire - 2) * 1_000L : 300;
        redisLockRenewalStrategy.renewal(key, () -> {
            //刷新过期时间，如果没有成功，可能是key不存在了，直接结束此任务
            if (!ObjectUtils.defaultIfNull(STRING_REDIS_TEMPLATE_LAZY.get().expire(key, expire, TimeUnit.SECONDS),
                    Boolean.FALSE)) {
                cancelScheduleIfRenewal();
            }
        }, period / 3, period, TimeUnit.MILLISECONDS);
    }

    private void cancelScheduleIfRenewal() {
        redisLockRenewalStrategy.cancel(key);
    }

    /**
     * 在指定的时间段获取锁，超出指定的时间立即返回
     *
     * @param time time
     * @param unit unit
     * @return true or false
     */
    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        if (Thread.interrupted()) {
            throw new InterruptedException();
        }
        long max = System.nanoTime() + unit.toNanos(time);
        while (System.nanoTime() < max) {
            if (tryLock()) {
                return true;
            }
        }
        return false;
    }

    /**
     * 解锁
     */
    @Override
    public void unlock() {
        STRING_REDIS_TEMPLATE_LAZY.get().execute(LUA_SCRIPT, ArrayUtils.asArrayList(key), value);
        if (renewal) {
            cancelScheduleIfRenewal();
        }
    }

    @Override
    public Condition newCondition() {
        throw new UnsupportedOperationException("Redis Lock newCondition");
    }

    public static void invoke(String key, Runnable runnable) {
        RedisLock.invoke(key, EXPIRE_SECONDS, runnable);
    }

    public static void invoke(String key, long expire, Runnable runnable) {
        RedisLock lock = new RedisLock(key, expire);
        lock.lock();
        try {
            runnable.run();
        } finally {
            lock.unlock();
        }
    }

    public static <T> T invoke(String key, Callable<T> callable) {
        return RedisLock.invoke(key, EXPIRE_SECONDS, callable);
    }

    @SneakyThrows
    public static <T> T invoke(String key, long expire, Callable<T> callable) {
        RedisLock lock = new RedisLock(key, expire);
        lock.lock();
        try {
            return callable.call();
        } finally {
            lock.unlock();
        }
    }
}
